\documentclass[cn,11pt,chinese]{elegantbook}

\usepackage{minted}
\usepackage{minted,tabularx,tikz}
\usepackage{graphicx}
\usetikzlibrary{
  shapes,
  shapes.geometric,
  decorations.text,
  shapes.geometric,
  calc,
  decorations.pathreplacing,
  automata,
  positioning,
  arrows
}

\usepackage{tcolorbox}
\tcbuselibrary{skins, breakable, theorems}

\usepackage{listings}
\usepackage{anyfontsize}

\definecolor{codegreen}{rgb}{0,0.6,0}
\definecolor{codegray}{rgb}{0.5,0.5,0.5}
\definecolor{codepurple}{rgb}{0.58,0,0.82}
\definecolor{backcolour}{rgb}{0.95,0.95,0.92}

\lstdefinestyle{mystyle}{
    backgroundcolor=\color{backcolour},   
    commentstyle=\color{codegreen},
    keywordstyle=\color{magenta},
    numberstyle=\tiny\color{codegray},
    stringstyle=\color{codepurple},
    basicstyle=\ttfamily\footnotesize,
    breakatwhitespace=false,         
    breaklines=true,                 
    captionpos=b,                    
    keepspaces=true,                 
    numbers=left,                    
    numbersep=5pt,                  
    showspaces=false,                
    showstringspaces=false,
    showtabs=false,                  
    tabsize=2
}

\lstset{style=mystyle}

\setcounter{tocdepth}{2}

\title{尚硅谷Flink教程}

% 本文档命令
\usepackage{array}
\newcommand{\ccr}[1]{\makecell{{\color{#1}\rule{1cm}{1cm}}}}

\begin{document}

\pagestyle{empty}

\begin{tikzpicture}[remember picture,overlay]
%%%%%%%%%%%%%%%%%%%% Background %%%%%%%%%%%%%%%%%%%%%%%%
\fill[Dandelion] (current page.south west) rectangle (current page.north east);




%%%%%%%%%%%%%%%%%%%% Background Polygon %%%%%%%%%%%%%%%%%%%%

\foreach \i in {2.5,...,22}
{
    \node[rounded corners,Dandelion!60,draw,regular polygon,regular polygon sides=6, minimum size=\i cm,ultra thick] at ($(current page.west)+(2.5,-5)$) {} ;
}

\foreach \i in {0.5,...,22}
{
\node[rounded corners,Dandelion!60,draw,regular polygon,regular polygon sides=6, minimum size=\i cm,ultra thick] at ($(current page.north west)+(2.5,0)$) {} ;
}

\foreach \i in {0.5,...,22}
{
\node[rounded corners,Dandelion!90,draw,regular polygon,regular polygon sides=6, minimum size=\i cm,ultra thick] at ($(current page.north east)+(0,-9.5)$) {} ;
}


\foreach \i in {21,...,6}
{
\node[Dandelion!85,rounded corners,draw,regular polygon,regular polygon sides=6, minimum size=\i cm,ultra thick] at ($(current page.south east)+(-0.2,-0.45)$) {} ;
}


%%%%%%%%%%%%%%%%%%%% Title of the Report %%%%%%%%%%%%%%%%%%%% 
\node[left,black,minimum width=0.625*\paperwidth,minimum height=3cm, rounded corners] at ($(current page.north east)+(0,-9.5)$)
{
{\fontsize{25}{30} \selectfont \bfseries Flink讲义}
};

%%%%%%%%%%%%%%%%%%%% Subtitle %%%%%%%%%%%%%%%%%%%% 
\node[left,black,minimum width=0.625*\paperwidth,minimum height=2cm, rounded corners] at ($(current page.north east)+(0,-11)$)
{
{\huge \textit{Flink重点内容概览}}
};

%%%%%%%%%%%%%%%%%%%% Author Name %%%%%%%%%%%%%%%%%%%% 
\node[left,black,minimum width=0.625*\paperwidth,minimum height=2cm, rounded corners] at ($(current page.north east)+(0,-13)$)
{
{\Large \textsc{左元}}
};

%%%%%%%%%%%%%%%%%%%% Year %%%%%%%%%%%%%%%%%%%% 
\node[rounded corners,fill=Dandelion!70,text =black,regular polygon,regular polygon sides=6, minimum size=2.5 cm,inner sep=0,ultra thick] at ($(current page.west)+(2.5,-5)$) {\LARGE \bfseries 2022};

\end{tikzpicture}

\frontmatter

\tableofcontents
%\listofchanges

\mainmatter

\chapter{Flink流处理简介}

\section{主要内容}

\begin{itemize}
  \item Flink是什么
  \item 为什么要用Flink
  \item 流处理的发展和演变
  \item Flink的主要特点
  \item Flink vs Spark Streaming
\end{itemize}

\section{Flink是什么}

Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行状态计算。

有界数据流

Flink会将文件看成一个流，一条一条的处理文件中的数据，例如下面包含两条数据的文本文件，将会一条一条处理，而不会整个文件进行处理。

\begin{minted}{text}
      Mary,./home,1970-01-01 00:00:01
      Bob,./cart,1970-01-01 00:00:02
\end{minted}

Flink Kafka Consumer，无界数据流，来一条处理一条

Flink底层引擎对于文件和Kafka消息队列的处理逻辑是一致的。

流批统一，一套业务逻辑代码搞定流和批。

\section{Flink目前在国内企业的应用}

所有大厂都在重度使用Flink。特别是阿里，收购了Flink母公司，并为Flink贡献了海量的代码，双十一大屏的指标全部使用Flink计算。几年前的数据：阿里巴巴每秒钟使用Flink处理4.6PB的数据量。

快手：Flink集群有1500台机器

字节跳动：Apache Storm $\rightarrow$ Apache Flink

Flink对所有的传统的流处理框架是降维打击。

\section{为什么选择Flink}

\begin{itemize}
  \item 流数据更真实地反映了我们的生活方式（点击流）
  
	\begin{figure}[htbp]
    \tikzstyle{process} = [rectangle, minimum width=3cm, minimum height=1cm, text centered, draw=black, fill=orange!30]
    \centering
    \scalebox{.8}{
	\tikzstyle{arrow} = [thick,->,>=stealth]
    \begin{tikzpicture}[node distance = 2cm]
      \node (pro1) [process] {点击事件};
      \node (pro2) [process, right of = pro1, xshift=2cm] {日志};
      \node (pro3) [process, right of = pro2, xshift=4cm] {消息队列};
      \node (pro4) [process, right of = pro3, xshift=2cm] {Flink};
      \draw [arrow] (pro1) -- (pro2);
      \draw [arrow] (pro2) -- node[anchor=south] {Flume} (pro3);
      \draw [arrow] (pro3) -- (pro4);
    \end{tikzpicture}}
  \end{figure}
  \item 传统的数据架构是基于有限数据集的，将数据流人为的变成了离线数据来处理
  
	\begin{figure}[htbp]
    \tikzstyle{process} = [rectangle, minimum width=3cm, minimum height=1cm, text centered, draw=black, fill=orange!30]
    \centering
    \scalebox{.8}{
	\tikzstyle{arrow} = [thick,->,>=stealth]
    \begin{tikzpicture}[node distance = 2cm]
      \node (pro1) [process] {点击事件};
      \node (pro2) [process, right of = pro1, xshift=2cm] {日志};
      \node (pro3) [process, right of = pro2, xshift=4cm] {消息队列};
      \node (pro4) [process, right of = pro3, xshift=2cm] {Hive};
      \draw [arrow] (pro1) -- (pro2);
      \draw [arrow] (pro2) -- node[anchor=south] {Flume} (pro3);
      \draw [arrow] (pro3) -- (pro4);
    \end{tikzpicture}}
  \end{figure}
  \item 我们的目标
    \begin{itemize}
      \item 低延迟（Spark Streaming 的延迟是秒级（需要攒批然后计算），Flink延迟是毫秒级（Spark Streaming的千分之一，Flink是来一条数据就处理一条数据，没有攒批的过程），由于操作系统的时钟的精度是毫秒级，所以可以认为Flink是没有延迟的）
      \item 高吞吐
      \item 结果的准确性和良好的容错性（EXACTLY-ONCE，恰好处理一次，精准一次消费）
    \end{itemize}
\end{itemize}

\begin{tcolorbox}
  Spark Streaming开窗口的大小要求是500毫秒（0.5秒）的整数倍

  延迟的概念：数据到达的时间和得到计算结果的时间的间隔
  
  吞吐的概念：单位时间内能处理的数据量的大小
  
  良好的容错性：程序宕机再重启，计算结果还是正确的。
  
  结果的准确性：某个事件的发生时间是：2020-01-01 00:00:09，
  
  如果我们开一个 10 秒钟的滚动窗口，窗口
  
  窗口一：2020-01-01 00:00:00 $\sim$ 2020-01-01 00:00:10
  
  窗口二：2020-01-01 00:00:10 $\sim$ 2020-01-01 00:00:20
  
  事件由于网络延迟，到达Spark Streaming服务器的时间是：
  
  2020-01-01 00:00:11，被分配到了窗口二中，导致计算结果的不准确。Flink是没有这个问题的。
\end{tcolorbox}

\section{哪些行业需要处理流数据}

\begin{itemize}
  \item 所有行业都需要处理流数据，因为数据本质上是流的形式。
  \item 电商：计算PV（Page View，页面访问次数统计）、UV（Unique Visitor，独立访客统计），以及实时热门商品（每过 5 分钟计算一次过去 1 小时的PV最多的商品）等指标。
  \item 物联网（IoT）：温度传感器连续 1 秒钟温度上升的检测。
  \item 风控：连续三次登录失败的检测（爬虫行为，暴力破解用户名和密码），信用卡欺诈检测（连续两笔消费，第一笔消费小于1元，第二笔消费大于500元），超时未支付订单的检测（美团下订单，15 分钟未支付，将关闭订单），刷单行为（连续三个事件：登录 $\rightarrow$ 下订单 $\rightarrow$ 支付）
\end{itemize}

\section{传统数据处理架构}

\begin{itemize}
  \item OLTP（在线事务处理）
  \item OLAP（在线分析处理）
  \item LAMBDA架构
\end{itemize}

\subsection{OLTP}

用一个关系型数据库完成所有需求

优点：架构很简单，适合初创公司。

数据库管理员（DBA）负责优化数据库（分库分表，建索引）
\begin{figure}[htbp]
  \centering
  \scalebox{.5}{
  \begin{tikzpicture}[node distance = 2cm,squarednode/.style={rectangle, draw=black, very thick, minimum height=1cm, minimum width=5cm},squarednode1/.style={rectangle, minimum size=1cm},]
		\node[squarednode] (1) {后台管理系统};
		\node[squarednode] (2) [right of = 1, xshift=5cm] {订单系统};
		\node[squarednode] (3) [right of = 2, xshift=5cm] {网站应用};
		\node[squarednode1] (4) [above of = 1, xshift=-1cm] {管理操作};
		\node[squarednode1] (5) [right of = 4] {响应};
		\node[squarednode1] (6) [above of = 2, xshift=-1cm] {下订单};
		\node[squarednode1] (7) [right of = 6] {响应};
		\node[squarednode1] (8) [above of = 3, xshift=-1cm] {点击流};
		\node[squarednode1] (9) [right of = 8] {响应};
		
		\draw[->] (4) -- (1);
		\draw[->] (1) -- (5);
		
		\draw[->] (6) -- (2);
		\draw[->] (2) -- (7);
		
		\draw[->] (8) -- (3);
		\draw[->] (3) -- (9);
		
		\node [cylinder,draw=black,thick,aspect=0.5,minimum height=1cm,minimum width=1cm,shape border rotate=90] (10) [below of = 2, yshift=-3cm] {关系型数据库};
		
		\draw[<->] (1)--(10);
		\draw[<->] (2)--(10);
		\draw[<->] (3)--(10);
		
		\draw[dashed] (-4,-2) -- (18, -2);
		
		\node[squarednode1] at (-3,-1.5) {计算};
		\node[squarednode1] at (-3,-2.5) {存储};
		
	\end{tikzpicture}}
\end{figure}

\subsection{OLAP}

优点：解耦合

缺点：得到计算结果的延迟比较大，不够实时

关系型数据库：用户表，用户地址表，订单表

统计指标的计算：Hive

\begin{figure}[htbp]
  \tikzstyle{process} = [rectangle, minimum width=3cm, minimum height=1cm, text centered, draw=black, fill=orange!30]
\tikzstyle{arrow} = [thick,->,>=stealth]
\centering
\scalebox{.7}{
  \begin{tikzpicture}[node distance = 2cm]
    \node (pro1) [process] {数据流};
    \node (pro2) [process, right of = pro1, xshift=2cm] {数据库};
    \node (pro3) [process, right of = pro2, xshift=4cm] {数据仓库};
    \node (pro4) [process, right of = pro3, xshift=4cm] {报表};
    \draw [arrow] (pro1) -- (pro2);
    \draw [arrow] (pro2) -- node[anchor=south] {Sqoop} (pro3);
    \draw [arrow] (pro3) -- node[anchor=south] {HIVE SQL} (pro4);
  \end{tikzpicture}}
\end{figure}

\subsection{LAMBDA架构}

\begin{itemize}
  \item 用两套系统，同时保证低延迟和结果准确性
  \item 使用批处理框架（hive）保证结果的准确性
  \item 使用流处理框架（Spark Streaming）保证结果的低延迟
\end{itemize}

\begin{figure}[htbp]
  \centering
  \scalebox{.7}{
  \begin{tikzpicture}[trim left=2cm,node distance = 2cm,squarednode/.style={rectangle, draw=black, very thick, minimum height=1cm, minimum width=3cm},squarednode1/.style={rectangle, minimum size=1cm},]
		\node[squarednode] (1) [xshift=5cm]{批处理系统};
		\node[squarednode] (2) [below of = 1] {流处理系统};
		\node[squarednode] (3) at (0,-1) {数据流};
		\node[squarednode] (4) at (15,-1){数据库};
		
		\draw[->] (3) -- (1);
		\draw[->] (3) -- (2);
		\draw[->] (1) -- node[above] {离线校正计算结果} ++(4);
		\draw[->] (2) -- node[below] {实时产生计算结果} ++(4);
	\end{tikzpicture}}
\end{figure}

\begin{note}
  \begin{enumerate}
    \item 先把程序跑起来
    \item 保证程序的正确性
    \item 优化程序
  \end{enumerate}
\end{note}

例子：

\begin{tcolorbox}
  点击事件：Mary,./Home,1970-01-01 00:00:09

  统计1970-01-01 00:00:00 $\sim$ 1970-01-01 00:00:10窗口的pv指标

  由于存在网络延时，事件在11秒时到达服务器，Spark Streaming可能将数据分到窗口1970-01-01 00:00:10 $\sim$ 1970-01-01 00:00:20

  这样的话，0 $\sim$ 10的统计结果就会少了一条，所以我们可以在晚些时候（例如晚上12点，使用HIVE再来计算一次0 $\sim$ 10窗口的页面访问量，校正一下计算结果，写入数据库）。
  \begin{itemize}
    \item Spark Streaming：保证结果的低延迟，但结果可能不准确
    \item Hive：保证结果的准确性，但结果的延迟比较高
  \end{itemize}

  两套系统计算的指标是同一个指标，所以写的代码也很类似。带来了维护的困难，因为一套代码要写两遍。

  能不能写一份代码就搞定低延迟和结果准确性呢？

  能不能使用一套框架既保证低延迟又保证结果的准确性呢？

  答案就是：Flink。

\end{tcolorbox}

\begin{figure}
  \tikzstyle{process} = [rectangle, minimum width=3cm, minimum height=1cm, text centered, draw=black, fill=orange!30, align=center]
\tikzstyle{arrow} = [thick,->,>=stealth]
\centering
\scalebox{.7}{
  \begin{tikzpicture}[node distance = 2cm]
    \node (pro1) [process] {数据流};
    \node (pro2) [process, right of = pro1, xshift=4cm] {Flink\\既保证低延时\\又保证结果的正确性};
    \node (pro3) [process, right of = pro2, xshift=4cm] {数据库};
    \draw [arrow] (pro1) -- (pro2);
    \draw [arrow] (pro2) -- (pro3);
  \end{tikzpicture}}
\end{figure}

\begin{note}
  Do Not Repeat Yourself. DRY.
\end{note}

\section{大数据的发展}

\begin{enumerate}
  \item 谷歌三篇论文（2003年）：GFS $\rightarrow$ HDFS，MapReduce $\rightarrow$ Hadoop，Big Table $\rightarrow$ HBase
  \item Spark（2008年）、Spark Streaming（将计算场景从磁盘转移到了内存，百倍提升）
  \item 谷歌：Dataflow Model $\rightarrow$ Apache Flink（2015年底）
\end{enumerate}

\section{有状态的流处理}

状态：有些算子可以维护内部状态

流处理：来一条数据处理一条（不攒批）

示意图：来一条数据统计值加一

\begin{enumerate}
  \item 来了一条数据（圆形），触发Flink应用程序的执行
  \item 读取内部状态（统计值）
  \item 统计值加1
  \item 将统计值写回内部状态
  \item 输出统计值（正方形）
\end{enumerate}

从图中可以看出，Flink处理数据是没有落盘操作的。

只在保存检查点（本地状态）的时候落盘。

sum算子中只会维护一个累加器，数据到来更新完累加器之后，数据直接被丢弃掉了。每个key对应的逻辑分区都有一个sum算子。

\begin{figure}[htbp]
  \centering
  \scalebox{.7}{
    \begin{tikzpicture}[node distance = 2cm,squarednode/.style={rectangle, draw=black, very thick, minimum height=1cm, minimum width=5cm},squarednode1/.style={rectangle, minimum size=1cm},]
      \node[squarednode, minimum height=5cm] (1) {};
      \node[squarednode] (2) at (0,2) {Flink应用程序};
      \node [cylinder,draw=black,thick,aspect=0.5,minimum height=1cm,minimum width=1cm,shape border rotate=90] (10) (3) [below of = 2, yshift=-1.5cm] {内部状态};
      
      \draw[->,postaction={decorate,decoration={text along path,text align=center,text={写入}}}] (2) to [bend left=45] (3);
      \draw[->,postaction={decorate,decoration={text along path,text align=center,text={读取}}}] (3) to [bend left=45] (2);
      
      \node [cylinder,draw=black,thick,aspect=0.5,minimum height=1cm,minimum width=1cm,shape border rotate=90] (10) (4) [below of = 1, yshift=-3cm] {远程存储};
      
      \draw[->] (3) -- (4);
      
      \draw[dashed] (-8,-3) -- (8, -3);
      
      \node[squarednode1] at (-6,-3.5) {周期性保存检查点};
      
      \node[circle,draw=black] (5) [left of = 2, xshift=-2cm] {};
      \node[circle,draw=black] (6) [left of = 5] {};
      \node[circle,draw=black] (7) [left of = 6] {};
      
      \node[draw=black] (8) [right of = 2, xshift=2cm] {};
      \node[draw=black] (9) [right of = 8] {};
      \node[draw=black] (10) [right of = 9] {};
      
      \draw (7) -- (6) -- (5) -- (2) -- (8) -- (9) -- (10);
      
    \end{tikzpicture}
  }
\end{figure}

数据流就像流水线

\begin{itemize}
  \item 数据是流水线上的产品
  \item 算子是流水线上的工人
\end{itemize}

\section{流处理的演变}

\begin{itemize}
  \item Apache Storm：低延迟，吞吐量很差，计算结果也不能保证准确性
  \item Apache Spark Streaming：高吞吐，延迟比较高，计算结果也不能保证准确性
  \item Apache Flink：低延迟，高吞吐，时间正确/语义化窗口，计算结果的正确性（EXACTLY-ONCE）
\end{itemize}

\section{Flink的主要特点}

\begin{itemize}
  \item 事件驱动
  \item 基于流的世界观
  \item 分层API
  \item 支持事件时间（EventTime）和处理时间（ProcessingTime）语义
    \begin{itemize}
      \item 事件时间（逻辑时钟）：事件发生的时间，时间戳包含在事件里
      \item 处理时间（物理时钟）：事件到达服务器的机器时间
    \end{itemize}
  \item 精确一次（EXACTLY-ONCE）的状态一致性保证
  \item 低延迟，每秒处理数百万个事件，毫秒级延迟
  \item 与众多常用存储系统的连接（ES，HBase，MySQL，Redis…）
  \item 高可用（Zookeeper），动态扩展，实现7*24小时全天候运行
\end{itemize}

\section{事件驱动（Event Driven）}

\begin{itemize}
  \item 来一条数据就处理一次，每来一条数据就会驱动DAG中算子的运行，也可以看作数据在DAG里面流动。
  \begin{figure}[htbp]
    \centering
    \scalebox{.8}{
      \begin{tikzpicture}
        \filldraw[fill=white, draw=black] (2,4) rectangle (7.5,6.5) node[pos=.5,yshift=-0.75cm] {无状态算子};
        \filldraw[fill=white, draw=black] (4.5,0.5) rectangle (10.5,3.5) node[pos=.5,yshift=-0.75cm] {有状态算子};
        \filldraw[fill=white, draw=black] (0,5) rectangle (1,6) node[pos=.5] {事件};
        \filldraw[fill=white, draw=black] (3,5) rectangle (4,6) node[pos=.5] {Map};
        \filldraw[fill=white, draw=black] (5,5) rectangle (7,6) node[pos=.5] {Filter};
        \filldraw[fill=white, draw=black] (8,5) rectangle (9.5,6) node[pos=.5] {KeyBy};
        \filldraw[fill=white, draw=black] (8,2) rectangle (10,3) node[pos=.5] {Reduce};
        
        \node [cylinder,draw=black,thick,aspect=0.5,minimum height=1cm,minimum width=1cm,shape border rotate=90] at (5.5,2.25) {累加器};
        
        \draw[->] (8,3) to[out=150,in=30] node[below] {写入} (6,3);
        \draw[->] (6,2) to[out=-30,in=-150] node[above] {读取} (8,2);
        \draw[->] (1,5.5) -- (3,5.5);
        \draw[->] (4,5.5) -- (5,5.5);
        \draw[->] (7,5.5) -- (8,5.5);
        \draw[->] (9,5) -- (9,3);
      \end{tikzpicture}
    }
  \end{figure}
  \item 事件到达之后立即驱动MAP的运行，MAP处理完事件之后，将ETL后的数据发送给FILTER算子，就会立刻驱动FILTER算子的运行，依次类推。
  \item 由于Flink是有状态的流处理，所以可能会有算子会维护和操作内部状态，例如REDUCE算子。而MAP和FILTER是无状态的计算。
  \item 传统批处理示意图如下：
  \begin{figure}[htbp]
    \tikzstyle{process} = [rectangle, minimum width=3cm, minimum height=1cm, text centered, draw=black, fill=orange!30, align=center]
  \tikzstyle{arrow} = [thick,->,>=stealth]
  \centering
  \scalebox{.7}{
    \begin{tikzpicture}[node distance = 2cm]
      \node (pro1) [process] {事件};
      \node (pro2) [process, right of = pro1, xshift=2cm] {事件};
      \node (pro3) [process, right of = pro2, xshift=2cm] {事件};
      \node (pro4) [circle, draw=black, right of = pro3, xshift=2cm] {攒批};
      \node (pro5) [rectangle, draw=black, right of = pro4, xshift=2cm] {计算结果};
      \draw [arrow] (pro1) -- (pro2);
      \draw [arrow] (pro2) -- (pro3);
      \draw [arrow] (pro3) -- (pro4);
      \draw [arrow] (pro4) -- (pro5);
    \end{tikzpicture}}
  \end{figure}
\end{itemize}

\begin{tcolorbox}
\textbf{流处理}

来一条数据处理一条。

flatMap算子什么时候会被触发执行？当flatMap算子的输入到达的时候，触发执行。

sum算子什么时候触发执行呢？当输入数据到达的时候，触发执行。来一条输入数据就要触发一次sum算子的执行。

算子都是被动执行的，数据不来不执行。

这个特性叫做事件驱动，输入事件驱动算子的执行。

流水线上的工人，上游处理过的产品到达，触发工人的操作。
\end{tcolorbox}

\section{基于流的世界观}

\begin{itemize}
  \item 在Flink的世界观中，一切都是由流组成的，离线数据是有界的流；实时数据是一个没有界限的流：这就是所谓的有界流和无界流。
  \item 在Spark Streaming的世界观中，一切都是由批组成的，离线数据是一批数据；实时数据是无数个微小批次组成的数据。
  \item 流的世界观最重要的一点其实是在静态的离线数据上面加了一个维度：时间
  \item 这个观点来自爱因斯坦的狭义相对论，批处理类似牛顿力学（坐标系：x，y，z），流处理类似狭义相对论力学（坐标系：x，y，z，t）。
\end{itemize}

\section{分层API}

\begin{figure}[htbp]
  \centering
  \scalebox{.7}{
    \begin{tikzpicture}[node distance = 2cm,squarednode/.style={rectangle, draw=black, very thick, minimum height=1cm, minimum width=5cm},squarednode1/.style={rectangle, minimum size=1cm},]
      \node[squarednode] (1) {底层API};
      \node[squarednode] (2) [above of = 1] {DataStream API};
      \node[squarednode] (3) [above of = 2] {Table API};
      \node[squarednode] (4) [above of = 3] {SQL};
      
      \draw[->,thick] (-4,-0.5) -- node[left] {抽象化} (-4,6.5);
      
      \draw[<-,thick] (4,-0.5) -- node[right] {具体化} (4,6.5);
      
    \end{tikzpicture}
  }
\end{figure}

\begin{itemize}
  \item 越抽象，越容易使用，但无法实现复杂的需求。
  \item 越底层，越难掌握，但可以使用很复杂的需求。
  \item 我们的学习重点：DataStream API和底层API。
\end{itemize}

\section{Flink中最重要的三个核心概念}

我们在学习Flink时，只需要彻底理解下面三个概念，就能够很好的理解Flink的本质：

\begin{itemize}
  \item 时间语义：事件时间，逻辑时钟（水位线），语义化窗口
  \item 状态：分清有状态的算子和无状态的算子的区别
  \item 事件驱动：来一条数据处理一次，Flink中的算子的计算逻辑都是被动执行的。
\end{itemize}

\section{分布式系统重要概念}

\begin{itemize}
  \item 分区：物理分区和逻辑分区的区别是什么
    \begin{itemize}
      \item hadoop的物理分区：机器节点/容器
      \item hadoop的逻辑分区：每个key对应的分组/分区数据
      \item spark的物理分区：core
      \item spark的逻辑分区：每个key所对应的逻辑分区
      \item flink的物理分区：任务插槽
      \item flink的逻辑分区：每个key所对应的逻辑分区
      \item 同一个逻辑分区的数据，一定在同一个物理分区
      \item 相同key的数据一定在一个节点上
    \end{itemize}
  \item 时钟：物理时钟和逻辑时钟的区别是什么
    \begin{itemize}
      \item flink中的物理时钟：机器时间
      \item flink中的逻辑时钟：水位线
    \end{itemize}
  \item 同步执行和异步执行的区别
    \begin{itemize}
      \item 同步：执行顺序是确定的
      \item 异步：执行顺序是不确定的
        \begin{itemize}
          \item 多进程
          \item 多线程
          \item 单线程异步IO（事件驱动）
        \end{itemize}
    \end{itemize}
\end{itemize}

\section{Flink vs Spark Streaming}

\begin{itemize}
  \item 流 vs 微批
  \item 事件驱动 vs 非事件驱动
  \item 数据模型
    \begin{itemize}
      \item Spark：RDD，Spark Streaming 的 DStream 实际上也就是一组组小批数据 RDD 的集合。
      \item Flink 基本数据模型是数据流，以及事件（Event）序列（Integer、String、Long、POJO Class、Tuple）
    \end{itemize}
  \item 运行时架构
    \begin{itemize}
      \item Spark是批计算，将DAG划分为不同的Stage，一个Stage完成后才可以计算下一个Stage。
      \item Flink是标准的流执行模式，一个事件在一个节点处理完后可以直接发往下一个节点进行处理。
    \end{itemize}
  \item Spark Streaming的延迟是Flink的1000倍。
  \item Flink支持事件时间和处理时间，Spark Streaming只支持处理时间
  \item Flink支持会话窗口
\end{itemize}

\section{Flink程序的典型结构}

\begin{enumerate}
  \item 获取流执行环境
  \item 设置并行任务的数量
  \item 读取数据源
  \item 进行计算
  \item 输出
  \item 执行程序
\end{enumerate}

\chapter{Flink运行时架构}

\section{HDFS主从架构}

\begin{figure}[htbp]
  \centering
  \scalebox{.6}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (2,2) rectangle (4,4) node[pos=.5] {NameNode};
      \filldraw[fill=white, draw=black] (6,6) rectangle (8,8) node[pos=.5] {DataNode};
      \filldraw[fill=white, draw=black] (6,2) rectangle (8,4) node[pos=.5] {DataNode};
      \filldraw[fill=white, draw=black] (6,-2) rectangle (8,0) node[pos=.5] {DataNode};
      
      \draw[->] (4,3) -- (6,7);
      \draw[->] (4,3) -- (6,3);
      \draw[->] (4,3) -- (6,-1);
    \end{tikzpicture}
  }
\end{figure}

\section{Spark主从架构}

\begin{figure}[htbp]
  \centering
  \scalebox{.6}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (2,2) rectangle (4,4) node[pos=.5] {Driver};
      \filldraw[fill=white, draw=black] (6,6) rectangle (8,8) node[pos=.5] {Executor};
      \filldraw[fill=white, draw=black] (6,2) rectangle (8,4) node[pos=.5] {Executor};
      \filldraw[fill=white, draw=black] (6,-2) rectangle (8,0) node[pos=.5] {Executor};
      
      \draw[->] (4,3) -- (6,7);
      \draw[->] (4,3) -- (6,3);
      \draw[->] (4,3) -- (6,-1);
    \end{tikzpicture}
  }
\end{figure}

\section{Flink主从架构}

\begin{itemize}
  \item Flink 运行时由两种类型的进程组成：一个JobManager（作业管理器，Master进程）和一个或者多个TaskManager（任务管理器，Slave进程）。
  \item 典型的Master-Slave（主从）架构。
\end{itemize}

\begin{figure}[htbp]
  \centering
  \scalebox{.6}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (1,2) rectangle (4,4) node[pos=.5] {JobManager};
      \filldraw[fill=white, draw=black] (6,6) rectangle (9,8) node[pos=.5] {TaskManager};
      \filldraw[fill=white, draw=black] (6,2) rectangle (9,4) node[pos=.5] {TaskManager};
      \filldraw[fill=white, draw=black] (6,-2) rectangle (9,0) node[pos=.5] {TaskManager};
      
      \draw[->] (4,3) -- (6,7);
      \draw[->] (4,3) -- (6,3);
      \draw[->] (4,3) -- (6,-1);
    \end{tikzpicture}
  }
\end{figure}

\section{作业管理器}

作业管理器是一个 JVM 进程。进程中包含三类线程：

\begin{itemize}
  \item Flink的资源管理器（ResourceManager）：资源是任务插槽（Task Slot）
  \item 分发器（WebUI）：提交任务和监控集群和任务
  \item JobMaster（每个作业对应一个）：调度任务，将DAG部署到任务管理器
\end{itemize}

\begin{figure}[htbp]
  \centering
  \scalebox{.6}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (1,2) rectangle (4,4) node[pos=.5] {作业管理器};
      \filldraw[fill=white, draw=black] (6,6) rectangle (9,8) node[pos=.5] {资源管理器};
      \filldraw[fill=white, draw=black] (6,2) rectangle (9,4) node[pos=.5] {分发器};
      \filldraw[fill=white, draw=black] (6,-2) rectangle (9,0) node[pos=.5] {JobMaster};
      
      \draw[->] (4,3) -- (6,7);
      \draw[->] (4,3) -- (6,3);
      \draw[->] (4,3) -- (6,-1);
    \end{tikzpicture}
  }
  \caption{作业管理器的三种线程}
\end{figure}

JobMaster由于是每个作业对应一个，所以可能有多个JobMaster线程。

\section{任务管理器}

\begin{itemize}
  \item 任务插槽是Flink的最小计算单元。
  \item 任务插槽（Task Slot）：是一个物理分区  
  \item 不同的任务插槽就是不同的物理分区
  \item 每个任务插槽是一个内存分片
  \item 内存的本质：字节数组
\end{itemize}

\begin{note}
  hadoop中的物理分区是DataNode。
\end{note}

\section{任务插槽}

\section{并行度的设置}

从上到下，优先级升高

\begin{enumerate}
  \item 任务管理器的配置文件里面：flink-conf.yaml中的配置项parallelism.default: 1
  \item 在命令行提交任务时指定并行度：./bin/flink run jar包 -p 16
  \item 全局并行度：env.setParallelism(1)
  \item 针对算子设置并行度：.print().setParallelism(1)
\end{enumerate}

\section{并行度设置的最佳实践}

\begin{enumerate}
  \item 不要设置全局并行度，因为没办法在命令行做动态扩容
  \item 针对某些算子设置并行度，例如数据源，为了不改变数据的顺序
  \item 在命令行设置，可以动态扩容
\end{enumerate}

\section{任务提交流程}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (0,0) rectangle (6,3) node[pos=.5] {作业管理器};
      \filldraw[fill=white, draw=black] (0,2) rectangle (1.5,3) node[pos=.5] {\tiny{分发器}};
      \filldraw[fill=white, draw=black] (4.5,0) rectangle (6,1) node[pos=.5] {\tiny{资源管理器}};
      \filldraw[fill=white, draw=black] (4.5,2) rectangle (6,3) node[pos=.5] {\tiny{Job Master}};
      
      \filldraw[fill=white, draw=black] (8,4) rectangle (10,5) node[pos=.5] {\tiny{任务管理器}};
      \filldraw[fill=white, draw=black] (8.5,4.5) rectangle (10.5,5.5) node[pos=.5] {\tiny{任务管理器}};
      \filldraw[fill=white, draw=black] (9,5) rectangle (11,6) node[pos=.5] {\tiny{任务管理器}};
      
      \filldraw[fill=white, draw=black] (-1,5) rectangle (1,6) node[pos=.5] {\tiny{Flink应用程序}};
      
      \draw[->] (0,5) -- node[left] {\tiny{1. 提交应用程序}} (0.75,3);
      \draw[->] (0.75,3) to[out=30,in=150] node[above] {\tiny{2. 启动Job Master}} (5.25,3);
      \draw[->] (6,2.5) to[out=-30,in=30] node[right] {\tiny{3. 请求任务插槽}} (6,0.5);
      \draw[->] (6,0.5) to[out=-30,in=-30] node[left,yshift=1cm,xshift=1cm] {\tiny{4. 发出请求任务插槽的指令}} (10, 4.5);
      \draw[->] (8,4.5) -- node[below,yshift=-0.2cm] {\tiny{5. 提供任务插槽}}(5.25,3);
      \draw[->] (5.25,3) to[out=120,in=120] node[above] {\tiny{6. 执行任务}} (8.5,5.5);
      
    \end{tikzpicture}
  }
  \caption{Flink任务提交流程}
\end{figure}

\begin{enumerate}
  \item 在集群启动时，任务管理器会向资源管理器注册自己的任务插槽
  \item 任务管理器之间存在数据的交换
\end{enumerate}

\section{Flink中的DAG数据结构}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (2,6) rectangle (5,7) node[pos=.5] {StreamGraph};
      \filldraw[fill=white, draw=black] (2,4) rectangle (5,5) node[pos=.5] {JobGraph};
      \filldraw[fill=white, draw=black] (2,2) rectangle (5,3) node[pos=.5] {ExecutionGraph};
      \filldraw[fill=white, draw=black] (2,0) rectangle (5,1) node[pos=.5] {物理执行图};
      
      \draw[->] (3.5,6) -- node[right] {客户端} (3.5,5);
      \draw[->] (3.5,4) -- node[right] {作业管理器} (3.5,3);
      \draw[->] (3.5,2) -- node[right] {任务管理器} (3.5,1);
    \end{tikzpicture}
  }
\end{figure}

\begin{itemize}
  \item StreamGraph：是根据用户通过Stream API编写的代码生成的最初的图。用来表示程序的拓扑结构。
  \item JobGraph：StreamGraph在编译的阶段经过优化后生成了JobGraph，提交给 JobManager的数据结构。主要的优化为，将多个符合条件（窄依赖，没有shuffle，并行度相同）的算子串在一起作为一个节点。保证串起来的节点里面的所有算子都在同一个任务插槽执行。这样算子之间的数据就是本地转发（无需序列化反序列化和网络IO）两个条件：
    \begin{itemize}
      \item 两个算子之间没有shuffle存在
      \item 两个算子的并行度必须相同
    \end{itemize}
  \item ExecutionGraph：JobManager根据JobGraph生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。
  \item 物理执行图：JobManager根据ExecutionGraph对Job进行调度后，在各个 TaskManager上部署Task后形成的“图”，并不是一个具体的数据结构。
\end{itemize}

\chapter{DataStream API}

\section{自定义数据源}

\subsection{POJO CLASS}

\begin{enumerate}
  \item 必须是公有类
  \item 所有字段必须是公有字段
  \item 必须有空构造器
\end{enumerate}

\subsection{SourceFunction}

SourceFunction<T>的泛型是数据源中的数据的类型。

\begin{itemize}
  \item run方法用来发送数据
  \item cancel方法在取消任务时执行
\end{itemize}

\begin{note}
  编写Flink程序时，要注意泛型和方法的参数类型！
\end{note}

\section{基本转换算子}

基本转换算子都是\textbf{无状态算子}。

举个无状态函数的例子

\begin{minted}{c}
      int add(int n) {
        return n + 1;
      }
\end{minted}

在输入相同的情况下，输出一定相同。

\begin{itemize}
  \item map的语义：针对流或者列表中的每一个元素，输出一个元素
  \item flatMap的语义：针对流或者列表中的每个元素，输出0个、1个或者多个元素
  \item filter的语义：针对流或者列表中的每个元素，输出0个或者1个元素
\end{itemize}

flatMap是map和filter的泛化，也就是说可以使用flatMap来实现map和filter的功能。

\section{逻辑分区算子}

\begin{itemize}
  \item keyBy：使用key将数据shuffle到不同的逻辑分区中
  \item 相同key的数据一定在同一个任务插槽（物理分区）中
  \item 不同key的数据也可能在同一个任务插槽（物理分区）中
  \item 同一个逻辑分区的数据一定在同一个物理分区中。
\end{itemize}

\begin{note}
  keyBy不是一个算子，因为不具备计算功能。keyBy的作用只是为数据指定key，并将数据路由到对应的逻辑分区。
\end{note}

reduce是有状态算子。举个有状态函数的例子。

\begin{minted}{c}
      int count = 0;
      int add(int n) {
        count += n;
        return count;
      }
\end{minted}

输入相同的情况下，输出不一定相同。

\begin{itemize}
  \item reduce的语义：reduce会初始化一个空累加器（类型和流中的元素类型相同），第一条元素到来，直接作为累加器保存，并将累加器输出。第二条以及之后的元素到来，和累加器进行累加操作并更新累加器，然后将累加器输出。reduce函数定义的是输入元素和累加器的累加规则。
  \item 每个key都会维护自己的累加器，输入数据更新完累加器之后，直接被丢弃
\end{itemize}

\subsection{reduce算子如何维护逻辑分区}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
      {HashMap};
      \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {key-1};
      \filldraw[fill=white, draw=black] (4,4) rectangle (7,5) node[pos=.5] {key-1的累加器};
      \draw[->] (3,4.5) -- (4,4.5);
      
      \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {key-2};
      \filldraw[fill=white, draw=black] (4,2) rectangle (7,3) node[pos=.5] {key-2的累加器};
      \draw[->] (3,2.5) -- (4,2.5);

      \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {reduce算子中的逻辑分区};
    \end{tikzpicture}
  }
\end{figure}

\section{物理分区算子}

将数据分发到不同的任务插槽（物理分区）

\begin{itemize}
  \item shuffle()
  \item rebalance()
  \item rescale()
  \item broadcast()
  \item global()
  \item custom()
\end{itemize}

\section{富函数}

生命周期（对象的生命周期，html页面的生命周期，锁的生命周期，数据库连接的生命周期，...）

算子的每一个并行子任务（任务槽）都有自己的生命周期。

\begin{itemize}
  \item open方法：在算子的计算逻辑执行前执行一次，适合做一些初始化的工作（打开一个文件，打开一个网络连接，打开一个数据库的连接）
  \item close方法：在算子的计算逻辑执行完毕之后执行一次，适合做一些清理工作。（关闭一个文件，关闭网络连接，关闭数据库连接）
  \item getRuntimeContext()方法：用来获取算子运行时的一些上下文信息。比如当前算子所处的并行子任务的索引等等。
\end{itemize}

举一些例子

MapFunction $\rightarrow$ RichMapFunction

FilterFunction $\rightarrow$ RichFilterFunction

FlatMapFunction $\rightarrow$ RichFlatMapFunction

SourceFunction $\rightarrow$ RichSourceFunction

SinkFunction $\rightarrow$ RichSinkFunction

\section{自定义输出}

SinkFunction<T>：泛型是要输出的数据的泛型

\chapter{底层API}

底层API（处理函数）都是富函数。

\section{ProcessFunction}

针对没有keyBy的数据流，可以使用ProcessFunction接口，针对流中的每个元素输出0个、1个或者多个元素。（非常类似RichFlatMapFunction）

\begin{itemize}
  \item ProcessFunction<IN, OUT>：IN是输入的泛型，OUT是输出的泛型
  \item processElement：每来一条数据，调用一次
  \item 使用.process(new ProcessFunction<I, O>)来调用。
\end{itemize}

\section{KeyedProcessFunction}

针对keyBy之后的键控流（KeyedStream），可以使用KeyedProcessFunction

\begin{itemize}
  \item KeyedProcessFunction<KEY, IN, OUT>：KEY是key的泛型，IN是输入的泛型，OUT是输出的泛型。
  \item processElement：来一条数据，触发调用一次。
  \item onTimer：定时器。时间到达某一个时间戳触发调用。
  \begin{figure}[htbp]
    \centering
    \scalebox{.8}{
      \begin{tikzpicture}[scale=1]
        \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
        (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
        {HashMap};
        \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {key-1};
        \filldraw[fill=white, draw=black] (4,4) rectangle (10,5) node[pos=.5] {key-1的定时器队列（优先队列）};
        \draw[->] (3,4.5) -- (4,4.5);
        
        \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {key-2};
        \filldraw[fill=white, draw=black] (4,2) rectangle (10,3) node[pos=.5] {key-2的定时器队列（优先队列）};
        \draw[->] (3,2.5) -- (4,2.5);
  
        \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {KeyedProcessFunction中的逻辑分区};
      \end{tikzpicture}
    }
  \end{figure}
  \item 每个key都会维护自己的定时器，每个key都只能访问自己的定时器。就好像每个key都只能访问自己的累加器一样。
  \item 针对每个key，在某个时间戳只能注册一个定时器，定时器不能重复注册，如果某个时间戳已经注册了定时器，那么再对这个时间戳注册定时器就不起作用了。
  \item .registerProcessingTimeTimer(ts)：在机器时间戳ts注册了一个定时器（onTimer）。
  \item 维护的内部状态
    \begin{itemize}
      \item 状态变量
      \item 定时器
    \end{itemize}
  \item processElement方法和onTimer方法：这两个方法是原子性的，无法并发执行。某个时刻只能执行一个方法。因为这两个方法都有可能操作相同的状态变量。例如：到达了一个事件，此时onTimer正在执行，则必须等待onTimer执行完以后，再调用processElement。再比如：到达了一个水位线，想触发onTimer，但此时processElement正在执行，那么必须等待processElement执行完以后再执行onTimer。
  \item 当水位线到达KeyedProcessFunction，如果这条水位线触发了onTimer的执行，则必须等待onTimer执行完以后，水位线才能向下游发送。
  \item 当水位线到达ProcessWindowFunction，如果这条水位线触发了process方法的执行，则必须等待process方法执行完以后，水位线才能向下游发送。
\end{itemize}

\begin{note}
  在KeyedProcessFunction中，可以认为维护了多张HashMap，每个状态变量的定义都会初始化一张HashMap，同时还有一张维护每个key的定时器队列的HashMap。
\end{note}

\section{逻辑分区维护的状态-键控状态变量}

每个key都会维护自己的状态变量

\begin{itemize}
  \item ValueState：类似Java的普通变量一样使用。
  \item ListState：类似Java的ArrayList一样使用。
  \item MapState：类似Java的HashMap一样使用。
\end{itemize}

\subsection{ValueState-值状态变量}

\begin{enumerate}
  \item 每个key都只能访问自己的状态变量，状态变量是每个key独有的。
  \item 状态变量是单例，只能被初始化一次。
  \item 状态变量会每隔一段时间作为检查点保存到状态后端（例如HDFS）。
  \item 当Flink程序启动时，会先去状态后端（例如HDFS）寻找状态变量，如果找不到，则初始化。如果找到了，则直接读取。所以说是单例。为什么Flink程序启动的时候，先去状态后端寻找状态变量呢？因为Flink不知道程序是第一次启动，还是故障恢复启动。如果是故障恢复，则要去保存的检查点里寻找状态变量，恢复到最近一次检查点的状态。
  \item .getState方法通过状态描述符去状态后端寻找状态变量
  \item 读取值状态变量中的值：.value()方法
  \item 将值写入状态变量：.update()方法
  \item 如何清空状态变量：.clear()方法
\end{enumerate}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
      {HashMap};
      \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {key-1};
      \filldraw[fill=white, draw=black] (4,4) rectangle (8,5) node[pos=.5] {key-1的ValueState};
      \draw[->] (3,4.5) -- (4,4.5);
      
      \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {key-2};
      \filldraw[fill=white, draw=black] (4,2) rectangle (8,3) node[pos=.5] {key-2的ValueState};
      \draw[->] (3,2.5) -- (4,2.5);

      \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {KeyedProcessFunction中的逻辑分区};
    \end{tikzpicture}
  }
\end{figure}

\subsection{ListState-列表状态变量}

\begin{itemize}
  \item .get()方法：返回包含列表状态变量中所有元素的迭代器
  \item .clear()方法：清空状态变量
  \item .add()方法：添加元素
\end{itemize}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
      {HashMap};
      \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {key-1};
      \filldraw[fill=white, draw=black] (4,4) rectangle (8,5) node[pos=.5] {key-1的ArrayList<T>};
      \draw[->] (3,4.5) -- (4,4.5);
      
      \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {key-2};
      \filldraw[fill=white, draw=black] (4,2) rectangle (8,3) node[pos=.5] {key-2的ArrayList<T>};
      \draw[->] (3,2.5) -- (4,2.5);

      \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {KeyedProcessFunction中的逻辑分区};
    \end{tikzpicture}
  }
\end{figure}

\subsection{MapState-字典状态变量}

\begin{itemize}
  \item .put(KEY, VALUE)方法：添加KEY $\rightarrow$ VALUE键值对
  \item .get(KEY)方法：获取KEY的VALUE
  \item .contains(KEY)方法：检测KEY是否存在
  \item .keys()：返回所有KEY组成的集合
\end{itemize}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
      {HashMap};
      \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {Mary};
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (4.5,3.5) -- (4.5,5.0) node [black,midway,xshift=-1.5cm] 
      {};
      \draw[->] (3,4.5) -- (4,4.5);
      \filldraw[fill=white, draw=black] (4.5,4.75) rectangle (6,5.25) node[pos=.5] {./home};
      \filldraw[fill=white, draw=black] (4.5,3.5) rectangle (6,4) node[pos=.5] {./cart};
      \filldraw[fill=white, draw=black] (7.5,4.75) rectangle (8,5.25) node[pos=.5] {2};
      \filldraw[fill=white, draw=black] (7.5,3.5) rectangle (8,4) node[pos=.5] {3};
      
      \draw[->] (6,5) -- (7.5,5);
      \draw[->] (6,3.75) -- (7.5,3.75);
      
      \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {John};
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (4.5,1.5) -- (4.5,3.0) node [black,midway,xshift=-1.5cm] 
      {};
      \draw[->] (3,2.5) -- (4,2.5);
      
      \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {.keyBy(r -> r.username).process中process的逻辑分区维护};
    \end{tikzpicture}
  }
\end{figure}

\section{ProcessWindowFunction}

使用在stream.keyBy().window()之后的流

\begin{itemize}
  \item ProcessWindowFunction<IN, OUT, KEY, WINDOW>
  \item process方法：窗口闭合的时候触发调用
  \item Flink中的窗口是左闭右开：[0,10)，当时间到达9999ms的时候，触发process函数的调用。
  \item 窗口从1970-01-01 00:00:00开始对齐
  \item 滚动窗口的计算公式，开了一个10秒钟的滚动窗口，7秒钟到达的事件属于哪个窗口？
  \item 窗口开始时间 = 时间戳 - 时间戳 \% 窗口大小
  \item 属于某个窗口的第一条数据到达以后才会开窗口
  \item 窗口内部状态：
    \begin{itemize}
      \item 属于窗口的所有事件
      \item 定时器：时间戳=窗口结束时间 - 1毫秒（因为是左闭右开区间），方法是process函数
    \end{itemize}
\end{itemize}

\begin{tcolorbox}
  process方法的迭代器参数包含了属于窗口的所有数据，会对内存造成压力，那么应该怎么去优化呢？使用累加器的思想。
\end{tcolorbox}

\subsection{AggregateFunction}

增量聚合函数，关键思想是在每个窗口中维护一个累加器。

\begin{itemize}
  \item AggregateFunction<IN, ACC, OUT>
  \item createAccumulator：创建空累加器，返回值的泛型是累加器的泛型
  \item add：定义输入数据和累加器的聚合规则，返回值是聚合后的累加器
  \item getResult：窗口闭合时发出聚合结果，返回值是将要发送的聚合结果
\end{itemize}

\subsection{将AggregateFunction和ProcessWindowFunction结合使用}

当窗口闭合时：增量聚合函数将getResult()方法的返回值，发送给了全窗口聚合函数

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}
      \filldraw[fill=white, draw=black] (-1,2) rectangle (3,3) node[pos=.5] {AggregateFunction};
      \filldraw[fill=white, draw=black] (6,2) rectangle (10,3) node[pos=.5] {ProcessWindowFunction};
      \draw[->] (3,2.5) -- node[above] {聚合结果} (6,2.5);
    \end{tikzpicture}
  }
\end{figure}

全窗口聚合函数中的迭代器中只有一个元素

\section{窗口的底层实现}

\subsection{增量聚合函数和全窗口聚合函数结合使用的底层实现}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
      {HashMap};
      \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {./home};
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (4.5,3.5) -- (4.5,5.0) node [black,midway,xshift=-1.5cm] 
      {};
      \draw[->] (3,4.5) -- (4,4.5);
      \filldraw[fill=white, draw=black] (4.5,4.75) rectangle (6,5.25) node[pos=.5] {0 $\sim$ 5};
      \filldraw[fill=white, draw=black] (4.5,3.5) rectangle (6,4) node[pos=.5] {5 $\sim$ 10};
      \filldraw[fill=white, draw=black] (6.5,4.75) rectangle (10,5.25) node[pos=.5] {0 $\sim$ 5的累加器};
      \filldraw[fill=white, draw=black] (6.5,3.5) rectangle (10,4) node[pos=.5] {5 $\sim$ 10的累加器};
      
      \draw[->] (6,5) -- (6.5,5);
      \draw[->] (6,3.75) -- (6.5,3.75);
      
      \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {./cart};
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (4.5,1.5) -- (4.5,3.0) node [black,midway,xshift=-1.5cm] 
      {};
      \draw[->] (3,2.5) -- (4,2.5);
      
      \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {增量聚合函数和全窗口聚合函数结合使用的底层实现};
    \end{tikzpicture}
  }
\end{figure}

\subsection{全窗口聚合函数的底层实现}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (0.5,1.5) -- (0.5,5.0) node [black,midway,xshift=-1.5cm] 
      {HashMap};
      \filldraw[fill=white, draw=black] (1,4) rectangle (3,5) node[pos=.5] {./home};
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (4.5,3.5) -- (4.5,5.0) node [black,midway,xshift=-1.5cm] 
      {};
      \draw[->] (3,4.5) -- (4,4.5);
      \filldraw[fill=white, draw=black] (4.5,4.75) rectangle (6,5.25) node[pos=.5] {0 $\sim$ 5};
      \filldraw[fill=white, draw=black] (4.5,3.5) rectangle (6,4) node[pos=.5] {5 $\sim$ 10};
      \filldraw[fill=white, draw=black] (6.5,4.75) rectangle (10,5.25) node[pos=.5] {0 $\sim$ 5的所有元素};
      \filldraw[fill=white, draw=black] (6.5,3.5) rectangle (10,4) node[pos=.5] {5 $\sim$ 10的所有元素};
      
      \draw[->] (6,5) -- (6.5,5);
      \draw[->] (6,3.75) -- (6.5,3.75);
      
      \filldraw[fill=white, draw=black] (1,2) rectangle (3,3) node[pos=.5] {./cart};
      \draw [decorate,decoration={brace,amplitude=10pt},xshift=-4pt,yshift=6pt]
      (4.5,1.5) -- (4.5,3.0) node [black,midway,xshift=-1.5cm] 
      {};
      \draw[->] (3,2.5) -- (4,2.5);
      
      \filldraw[fill=white,draw=white] (-1,0) rectangle (5,1) node[pos=.5] {全窗口聚合函数结合使用的底层实现};
    \end{tikzpicture}
  }
\end{figure}

窗口中的所有元素都保存在List中。

\section{ProcessAllWindowFunction}

\begin{minted}{java}
      stream
        .windowAll()
        .process(new ProcessAllWindowFunction())
\end{minted}

直接对流进行开窗口，等价于将所有数据keyBy到同一条流，然后进行开窗口

\begin{minted}{java}
      stream
        .keyBy(r -> 1)
        .window()
        .process(new ProcessWindowFunction())
\end{minted}

\chapter{窗口}

\section{窗口的本质}

我们一般在.keyBy之后使用.window方法来进行开窗，实际上是在keyBy之后的逻辑分区中，再按照窗口进行一次逻辑分区。先分流再开窗。

\begin{minted}{sql}
      SELECT * FROM table GROUP BY key, window;
\end{minted}

Flink窗口是左闭右开的区间，例如[0, 5)的窗口最后一个时间戳是4999毫秒。

\section{窗口概念}

\begin{itemize}
  \item 一般真实的流都是无界的，怎样处理无界的数据？
  \item 可以把无限的数据流进行切分，得到有限的数据集进行处理—也就是得到有界流
  \item 窗口（Window）就是将无限流切割为有限流的一种方式，它会将流数据分发到有限大小的桶（bucket）中进行分析
\end{itemize}

\section{窗口类型}

\begin{itemize}
  \item 时间窗口（Time Window）
    \begin{itemize}
      \item 滚动时间窗口
      \item 滑动时间窗口
      \item 会话窗口
    \end{itemize}
  \item 计数窗口（Count Window）
    \begin{itemize}
      \item 滚动计数窗口
      \item 滑动计数窗口
    \end{itemize}
\end{itemize}

\section{滚动窗口}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
	
	
      % window-1
      \filldraw[fill=white, draw=red,thick,dashed] (0,9) rectangle (4,11) node[pos=.5,yshift=1.5cm] {window-1};
      
      
      \draw (1.5,10) circle (0.5);
      \draw (2.5,10) circle (0.5);
      
      \filldraw[fill=white, draw=red,thick,dashed] (0,5) rectangle (4,7) node[pos=.5,yshift=1.5cm] {window-1};
      
      \draw (1.5,6) circle (0.5);
      \draw (2.5,6) circle (0.5);
      
      \filldraw[fill=white, draw=red,thick,dashed] (0,1) rectangle (4,3) node[pos=.5,yshift=1.5cm] {window-1};
      
      \draw (1.5,2) circle (0.5);
      \draw (2.5,2) circle (0.5);
      
      % window-2
      \filldraw[fill=white, draw=green,thick,dashed] (4,9) rectangle (8,11) node[pos=.5,yshift=1.5cm] {window-2};
      
      \draw (5.5,10) circle (0.5);
      \draw (6.5,10) circle (0.5);
      
      \filldraw[fill=white, draw=green,thick,dashed] (4,5) rectangle (8,7) node[pos=.5,yshift=1.5cm] {window-2};
      
      \draw (5.5,6) circle (0.5);
      \draw (6.5,6) circle (0.5);
      
      \filldraw[fill=white, draw=green,thick,dashed] (4,1) rectangle (8,3) node[pos=.5,yshift=1.5cm] {window-2};
      
      \draw (5.5,2) circle (0.5);
      \draw (6.5,2) circle (0.5);
      
      
      % window-3
      \filldraw[fill=white, draw=blue,thick,dashed] (8,9) rectangle (12,11) node[pos=.5,yshift=1.5cm] {window-3};
      
      \draw (9.5,10) circle (0.5);
      \draw (10.5,10) circle (0.5);
      
      \filldraw[fill=white, draw=blue,thick,dashed] (8,5) rectangle (12,7) node[pos=.5,yshift=1.5cm] {window-3};
      
      \draw (9.5,6) circle (0.5);
      \draw (10.5,6) circle (0.5);
      
      \filldraw[fill=white, draw=blue,thick,dashed] (8,1) rectangle (12,3) node[pos=.5,yshift=1.5cm] {window-3};
      
      \draw (9.5,2) circle (0.5);
      \draw (10.5,2) circle (0.5);
      
      % key
      \filldraw[fill=white, draw=white] (-2,10) rectangle (0,10.5) node[pos=.5] {user-1};
      \filldraw[fill=white, draw=white] (-2,6) rectangle (0,6.5) node[pos=.5] {user-2};
      \filldraw[fill=white, draw=white] (-2,2) rectangle (0,2.5) node[pos=.5] {user-3};
      
      % 坐标轴
      \draw[->,draw=black] (0,0) -- (12, 0);
      \draw[->,draw=black] (0,0) -- (0, 12);
      
      \end{tikzpicture}
  }
  \caption{滚动窗口}
\end{figure}

\begin{itemize}
  \item 将数据依据固定的窗口长度对数据进行切分
  \item 时间对齐，窗口长度固定，没有重叠
\end{itemize}

\section{滑动窗口}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{
    \begin{tikzpicture}[scale=1]
	
      % window-1
      \filldraw[fill=white, draw=red,thick,dashed] (0,9) rectangle (6,11) node[pos=.5,yshift=1.5cm] {window-1};
      
      
      \filldraw[fill=white, draw=red,thick,dashed] (0,5) rectangle (6,7) node[pos=.5,yshift=1.5cm] {window-1};
      
      \filldraw[fill=white, draw=red,thick,dashed] (0,1) rectangle (6,3) node[pos=.5,yshift=1.5cm] {window-1};
      
      
      
      % window-2
      \filldraw[fill=white, draw=green,thick,dashed] (4,8.9) rectangle (10,10.9) node[pos=.5,yshift=1.5cm] {window-2};
      
      \draw (5.5,10) circle (0.3);
      \draw (6.5,10) circle (0.3);
      
      \filldraw[fill=white, draw=green,thick,dashed] (4,4.9) rectangle (10,6.9) node[pos=.5,yshift=1.5cm] {window-2};
      
      \draw (5.5,6) circle (0.3);
      \draw (6.5,6) circle (0.3);
      
      \filldraw[fill=white, draw=green,thick,dashed] (4,0.9) rectangle (10,2.9) node[pos=.5,yshift=1.5cm] {window-2};
      
      \draw (5.5,2) circle (0.3);
      \draw (6.5,2) circle (0.3);
      
      
      % key
      \filldraw[fill=white, draw=white] (-2,10) rectangle (0,10.5) node[pos=.5] {user-1};
      \filldraw[fill=white, draw=white] (-2,6) rectangle (0,6.5) node[pos=.5] {user-2};
      \filldraw[fill=white, draw=white] (-2,2) rectangle (0,2.5) node[pos=.5] {user-3};
      
      % 坐标轴
      \draw[->,draw=black] (0,0) -- (12, 0);
      \draw[->,draw=black] (0,0) -- (0, 12);
      
      % window-1
      \draw (1.5,6) circle (0.3);
      \draw (4.5,6) circle (0.3);
      
      \draw (1.5,10) circle (0.3);
      \draw (4.5,10) circle (0.3);
      
      \draw (1.5,2) circle (0.3);
      \draw (4.5,2) circle (0.3);
      
      %hack
      \draw[draw=red,dashed,thick] (4,9) -- (6,9);
      \draw[draw=red,dashed,thick] (6,9) -- (6,10.9);
      
      \draw[draw=red,dashed,thick] (4,5) -- (6,5);
      \draw[draw=red,dashed,thick] (6,5) -- (6,6.9);
      
      \draw[draw=red,dashed,thick] (4,1) -- (6,1);
      \draw[draw=red,dashed,thick] (6,1) -- (6,2.9);
      
      \end{tikzpicture}
  }
  \caption{滑动窗口}
\end{figure}

\begin{itemize}
  \item 滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成
  \item 窗口长度固定，可以有重叠
  \item 如果一个元素同时属于两个窗口，那么会将元素复制两份，每个窗口分配一个元素。这样就可以保证所有的窗口从逻辑上分开处理。
\end{itemize}

\section{会话窗口}

\begin{figure}[htbp]
  \centering
  \scalebox{.8}{

  }
  \caption{会话窗口}
\end{figure}

\chapter{逻辑时钟-水位线}

由于事件时间的世界里面没有时钟，所以我们需要为这个世界提供时钟，叫做水位线（逻辑时钟）。

时钟的定义：单调递增的序列。

如果单调递增的序列是CPU产生的，那么就是物理时钟。如果是编程产生的，那么就是逻辑时钟。

水位线流动到哪一个算子，哪一个算子就会更新自己的时钟（产线工人的手表）。水位线不会弯道超车，也就是说，水位线不会绕过水位线前面的数据传播下去。如果某个算子的计算过程很耗时，那么会阻塞水位线的向下传播。如果上游的算子阻塞了水位线的传播，那么上游算子和下游算子的水位线（时钟）可能是不一样的。

在使用处理时间（ProcessingTime）的情况下，不存在乱序数据的情况。

乱序数据只存在于使用事件时间的情况下。

时钟：

\begin{itemize}
  \item 物理时钟：机器时间，墙上时钟
  \item 逻辑时钟：水位线
\end{itemize}

在Flink里面，时钟的作用是什么？

\begin{itemize}
  \item 触发定时器
  \item 关闭窗口
\end{itemize}

\section{有关水位线的一些约定}

\begin{theorem}{}{}
  水位线是事件时间的世界中的逻辑时钟。
\end{theorem}

\begin{theorem}{}{}
  Flink认为携带时间戳 $\le$ 水位线的事件都已经到达。
\end{theorem}

\begin{theorem}{}{}
  水位线是一种特殊的事件。由程序员编程插入的数据流中。随数据流流动。
\end{theorem}

\begin{theorem}{}{}
  算子的每个并行子任务会维护自己的逻辑时钟，当接收到上游发送过来的水位线时，更新自己的逻辑时钟。
\end{theorem}

\begin{theorem}{}{}
  水位线 = 观察到的最大时间戳 - 最大延迟时间 - 1 毫秒
\end{theorem}

\begin{theorem}{}{}
  \begin{itemize}
    \item Flink会在流的最开始插入一个时间戳为负无穷大的水位线
    \item Flink会在流的最末尾插入一个时间戳为正无穷大的水位线
  \end{itemize}
\end{theorem}

\begin{theorem}{}{}
  \begin{itemize}
    \item 水位线 $\ge$ 窗口结束时间，触发窗口计算。
    \item 水位线 $\ge$ 定时器时间戳，触发定时器执行。
  \end{itemize}
\end{theorem}

\section{水位线设置的最佳实践}

\begin{itemize}
  \item 在尽量靠近数据源的地方设置水位线（assignTimestampsAndWatermarks）
  \item assignTimestampsAndWatermarks前面的算子的并行度最好是1
\end{itemize}

\section{Kafka水位线设置}

数据源的并行度等于Kafka主题的分区数量。

插入水位线之前的所有算子的并行度和Kafka主题的分区数量一致。

\section{如何处理迟到元素}

到达的数据的时间戳小于当前算子的水位线，就是\textbf{迟到数据}。只有在使用事件时间的情况下，才会存在迟到元素。对于处理时间来说，不存在迟到数据。

\begin{enumerate}
  \item 默认策略：直接丢弃掉迟到数据。当数据到达以后，发现所属窗口已经闭合计算并销毁了，那么数据就被直接丢弃了。
    \begin{itemize}
      \item 如果来的数据是迟到数据，但是所属窗口还在，那么数据可以进入窗口。
      \item 如果来的数据是迟到数据，但所属窗口已经销毁，那么数据被丢弃。
    \end{itemize}
  \item 将迟到数据发送到侧输出流中去。侧输出流是不同于主流输出的旁路输出流，可以向侧输出流发送任意数据。
  \item 使用迟到元素更新窗口计算结果。也就是当水位线到达窗口结束时间的时候，触发窗口计算，但不销毁窗口，而是选择再等待迟到元素一段时间。
    \begin{itemize}
      \item .allowedLateness(Time.seconds(5))：窗口会等待5秒钟的迟到事件
      \item 窗口真正销毁：水位线 $\ge$ 窗口结束时间 + allowedLateness
      \item 窗口的第一次触发计算：水位线 $\ge$ 窗口结束时间，触发计算完以后窗口不会被销毁
    \end{itemize}
\end{enumerate}

处理迟到数据的作用：将迟到数据保存下来，然后统计一下每天有多少迟到数据，然后方便我们调整最大延迟时间的设置，改进计算结果的正确性。

\section{如何处理迟到数据}

到达的数据的时间戳小于当前算子的水位线，就是迟到数据。只有在使用事件时间的情况下，才会存在迟到元素。对于处理时间来说，不存在迟到数据。

处理迟到数据的策略：

\begin{enumerate}
  \item 默认策略：直接丢弃掉迟到数据。当数据到达以后，发现所属窗口已经闭合计算并销毁了，那么数据就被直接丢弃了。
    \begin{itemize}
      \item 如果来的数据是迟到数据，但是所属窗口还在，那么数据可以进入窗口。
      \item 如果来的数据是迟到数据，但所属窗口已经销毁，那么数据被丢弃。
    \end{itemize}
  \item 将迟到数据发送到侧输出流中去。侧输出流是不同于主流输出的旁路输出流，可以向侧输出流发送任意数据。
  \item 使用迟到元素更新窗口计算结果。也就是当水位线到达窗口结束时间的时候，触发窗口计算，但不销毁窗口，而是选择再等待迟到元素一段时间。
    \begin{itemize}
      \item .allowedLateness(Time.seconds(5))：窗口会等待5秒钟的迟到事件
      \item 窗口真正销毁：水位线 $\ge$ 窗口结束时间 + allowedLateness
      \item 窗口的第一次触发计算：水位线 $\ge$ 窗口结束时间，触发计算完以后窗口不会被销毁
    \end{itemize}
\end{enumerate}

\textbf{处理迟到数据的作用}：将迟到数据保存下来，然后统计一下每天有多少迟到数据，然后方便我们调整最大延迟时间的设置，改进计算结果的正确性。

\chapter{多流合并}

Flink的多流合并的机制是以FIFO的方式合并多条流。

\begin{itemize}
  \item union
    \begin{itemize}
      \item 多条流的元素类型必须一样
      \item 可以合并多条流
    \end{itemize}
  \item connect
    \begin{itemize}
      \item 只能合并两条流
      \item 两条流的元素的类型可以不一样
      \item CoMapFunction/CoFlatMapFunction
      \item CoProcessFunction
      \item 应用场景
        \begin{itemize}
          \item 一条流进行keyBy，另一条流broadcast，使的所有的逻辑分区都能和同一条流进行JOIN。
          \item 两条流都进行keyBy，将来自两条流的相同key的数据合并在一起处理，也就是说，将来自两条流的相同key的数据放在一个逻辑分区中做处理。
        \end{itemize}
    \end{itemize}
\end{itemize}

\chapter{Flink中的状态管理}

\section{状态后端}

\begin{itemize}
  \item 每传入一条数据，有状态的算子任务都会读取和更新状态
  \item 由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务都会在本地维护其状态，以确保快速的状态访问
  \item 状态的存储、访问以及维护，由一个可插入的组件决定，这个组件就叫做状态后端（state backend）
  \item 状态后端主要负责两件事：本地的状态管理，以及将检查点（checkpoint）状态写入远程存储（HDFS、RocksDB、文件系统之类的）
\end{itemize}

\section{选择一个状态后端}

\begin{itemize}
  \item MemoryStateBackend（默认）
    \begin{itemize}
      \item 内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager的JVM堆上，而将checkpoint存储在JobManager的内存中
      \item 特点：快速、低延迟，但不稳定
    \end{itemize}
  \item FsStateBackend
    \begin{itemize}
      \item 将checkpoint存到远程的持久化文件系统（FileSystem）上，而对于本地状态，跟MemoryStateBackend一样，也会存在TaskManager的JVM堆上
      \item 同时拥有内存级的本地访问速度，和更好的容错保证
    \end{itemize}
  \item RocksDBStateBackend
    \begin{itemize}
      \item 将所有状态序列化后，存入本地的RocksDB中存储。
      \item RocksDB是一个硬盘KV数据库。
    \end{itemize}
\end{itemize}

\chapter{Flink的容错机制}

\chapter{端到端一致性}

\chapter{背压问题}

背压问题的本质：单位时间缓冲区入队列的数据量，大于缓冲区出队列的数据量。

\begin{enumerate}
  \item 减慢入队列（生产者）的速度
  \item 加快出队列（消费者）的速度
  \item 增大缓冲区队列的大小
  \item 消费者告诉生产者自己目前能消费多少（信任度）
\end{enumerate}

下游算子会向上有算子发送一个消息，告诉上游算子自己的接收缓冲区能接收多少数据，当上游算子的缓冲区攒够 $\le$ 下游算子的接收缓冲区能接收的数据量，就发送下去。

例如：

下游算子的接收缓冲区的大小是4，但目前只有3个元素，也就是说还能接收1个元素，那么会发送一条消息给上游算子，告诉上游算子自己的缓冲区还有1个空位，这样当上游算子的发送缓冲区里有1条数据，直接发送给下游算子的接收缓冲区。

\textbf{信任度算法}

信任度：下游算子的接收缓冲区还能接收多少条数据。

\textbf{基于信用度的流量控制}

通过网络连接来发送每条数据的效率很低，会导致很大的开销。为了充分利用网络连接的带宽，就需要进行缓冲了。在流处理的上下文中，缓冲的一个缺点是会增加延迟，因为数据需要在缓冲区中进行收集，而不是立即发送。

Flink实现了一个基于信任度的流量控制机制，其工作原理如下。接收任务授予发送任务一些“信任度”（credit），也就是为了接收其数据而保留的网络缓冲区数。当发送者收到一个信任度通知，它就会按照被授予的信任度，发送尽可能多的缓冲数据，并且同时发送目前积压数据的大小——也就是已填满并准备发送的网络缓冲的数量。接收者用保留的缓冲区处理发来的数据，并对发送者传来的积压量进行综合考量，为其所有连接的发送者确定下一个信用度授权的优先级。

基于信用度的流控制可以减少延迟，因为发送者可以在接收者有足够的资源接受数据时立即发送数据。此外，在数据倾斜的情况下，这样分配网络资源是一种很有效的机制，因为信用度是根据发送者积压数据量的规模授予的。因此，基于信用的流量控制是Flink实现高吞吐量和低延迟的重要组成部分。

\chapter{有限状态机}

\chapter{竞争状态和锁机制}

\end{document}